package com.yanqu.manager;

import com.alibaba.fastjson.JSON;
import com.yanqu.data.entity.LogGoodsEntity;
import com.yanqu.road.utils.MyProperties;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.util.ResourceUtils;
import org.springframework.util.StringUtils;

import java.io.File;
import java.time.Duration;
import java.util.Collections;

/**
 * 道具获得阈值监控的消费者
 */
public class GoodsMonitorKafkaConsumer {

    private String topic = "dzg_get_goods_log";

    private static final Logger logger = LogManager.getLogger(GoodsMonitorKafkaConsumer.class.getName());

    private KafkaConsumer<String, String> kafkaConsumer;

    private volatile boolean isStop = false;

    public GoodsMonitorKafkaConsumer() {

        String configPath = "";
        try{
            configPath = ResourceUtils.getFile("").getAbsolutePath() + File.separator+ "config" +File.separator;
        }catch (Exception e){
            logger.error("", e);
            throw new RuntimeException(e);
        }

        MyProperties myProperties = new MyProperties(configPath + "kafka.properties");

        if(StringUtils.isEmpty(myProperties.getProperty("bootstrap.servers"))){
            logger.info("no kafka config !!!!!!!!!");
            return;
        }

        this.kafkaConsumer = new KafkaConsumer<>(myProperties.getProperties());


        Thread thread = new Thread(() -> {

            kafkaConsumer.subscribe(Collections.singleton(topic));

            while (!isStop) {

                try {

                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(800));
                    records.forEach(record -> {
                        LogGoodsEntity entity = JSON.parseObject(record.value(), LogGoodsEntity.class);
                        UserGetGoodsMonitorMgr.recordLogGoodsEntity(entity);
                    });

                } catch (Exception e) {
                    logger.error("", e);
                }
            }

            kafkaConsumer.close();

            logger.info("kafkaConsumer close");
        });
        thread.start();

    }

    public void close(){
        this.isStop = true;
    }

}
